{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "14f4149a",
   "metadata": {},
   "source": [
    "# Demo: Real-Time Feature Engineering with Feldera"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c2c57a97",
   "metadata": {},
   "source": [
    "## INTRODUCTION\n",
    "\n",
    "Feature engineering is the process of transforming raw data into features that better represent the underlying problem to an ML model. While raw data represents individual events, features summarize many raw events, e.g., average readings collected over a period of time. When raw input data arrives continuously in real-time, feature computation must also be performed in real-time in order to supply the ML model with up-to-date inputs. \n",
    "\n",
    "Building real-time feature pipelines using existing tools like Flink and Spark Structured Streaming is notoriously hard. **Feldera simplifies this task dramatically**. As a user, you simply express your features as SQL queries. Feldera runs these queries over historical data in order to compute feature vectors for model training and testing. Then, during inference, Feldera evaluates **the same queries** over streaming inputs in real-time.\n",
    "\n",
    "In this notebook, we use Feldera to build a real-time credit card fraud detector. In particular, we: \n",
    "* Write SQL queries that define several interesting features, based on data enrichment and rolling aggregates.\n",
    "* Compute feature vectors and train an ML model on a historical data set stored in a Delta Lake.\n",
    "* Use the same queries to compute feature vectors over a real-time stream of credit card transactions."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "edd7537c",
   "metadata": {},
   "source": [
    "## USE CASE: CREDIT CARD FRAUD DETECTION\n",
    "Credit card fraud detection is a classic application of real-time feature engineering. Here, data comes in a stream of **transactions**, each with attributes like card number, purchase time, vendor, and amount. Additionally, the fraud detector has access to a slowly changing table with **demographics** information about cardholders, such as age and address.\n",
    "\n",
    "<img src=\"./tables.png\" alt=\"training\" width=\"800\"/>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a8374892",
   "metadata": {},
   "source": [
    "## INPUT DATASETS\n",
    "We used a publicly available [Synthetic Credit Card Transaction Generator](https://github.com/namebrandon/Sparkov_Data_Generation) to generate two labeled datasets, both with 1000 user profiles.  We will use the first dataset for model training and testing, and the second dataset -- for real-time inference.  We stored the datasets in the [Delta Lake format](https://delta.io/) in two public S3 buckets:\n",
    "\n",
    "* Training dataset:\n",
    "  * Demographics table: `s3://feldera-fraud-detection-data/demographics_train/`\n",
    "  * Transaction table: `s3://feldera-fraud-detection-data/transaction_train/`\n",
    "\n",
    "* Inference dataset:\n",
    "  * Demographics table: `s3://feldera-fraud-detection-data/demographics_train/`\n",
    "  * Transaction table: `s3://feldera-fraud-detection-data/transaction_train/`\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0ae0dcd2",
   "metadata": {},
   "source": [
    "## MODEL TRAINING AND TESTING\n",
    "\n",
    "Finding an optimal set of features to train a good ML model is an iterative process. At every step, the data scientist trains and tests a model using currently selected feature queries on labeled historical data. The results of each experiment drive the next refinement of feature queries.\n",
    "\n",
    "<img src=\"training_animation.gif\" alt=\"training\" width=\"800\"/>\n",
    "\n",
    "Below we show one iteration of this process: we define a set of features, train a model using these features, and test its accuracy.\n",
    "\n",
    "### Feature queries\n",
    "\n",
    "We define several features over our input tables:\n",
    "\n",
    "* Data enrichment:\n",
    "  * We add demographic attributes, such as zip code, to each transaction\n",
    "* Rolling aggregates:\n",
    "  * average spending per transaction in the past day, week, and month\n",
    "  * average spending per transaction over a 3-month timeframe on the same day of the week\n",
    "  * number of transactions made with this credit card in the last 24 hours\n",
    "* Other:\n",
    "  * `is_weekend` - transaction took place on a weekend\n",
    "  * `is_night` - transaction took place before 6am\n",
    "  * `d` - day of week\n",
    "\n",
    "The following Python function uses the Feldera Python SDK to creates an SQL program, consisting of two tables with raw input data (`TRANSACTION` and `DEMOGRAPHICS`) and the `FEATURE` view, which computes the above features over these tables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e6724c8f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from feldera import FelderaClient, SQLContext, SQLSchema\n",
    "\n",
    "def build_program(client, pipeline_name):\n",
    "    sql = SQLContext(pipeline_name, client).get_or_create()\n",
    "    # Declare input table with raw credit card transaction data.\n",
    "    sql.register_table(\n",
    "        \"TRANSACTION\",\n",
    "        SQLSchema(\n",
    "            {\n",
    "                \"trans_date_trans_time\": \"TIMESTAMP\",\n",
    "                \"cc_num\": \"BIGINT\",\n",
    "                \"merchant\": \"STRING\",\n",
    "                \"category\": \"STRING\",\n",
    "                \"amt\": \"DOUBLE\",\n",
    "                \"trans_num\": \"STRING\",\n",
    "                \"unix_time\": \"BIGINT\",\n",
    "                \"merch_lat\": \"DOUBLE\",\n",
    "                \"merch_long\": \"DOUBLE\",\n",
    "                \"is_fraud\": \"BIGINT\",\n",
    "            }\n",
    "        ),\n",
    "    )\n",
    "\n",
    "    # Declare input table with demographics data.\n",
    "    sql.register_table(\n",
    "        \"DEMOGRAPHICS\",\n",
    "        SQLSchema(\n",
    "            {\n",
    "                \"cc_num\": \"BIGINT\",\n",
    "                \"first\": \"STRING\",\n",
    "                \"last\": \"STRING\",\n",
    "                \"gender\": \"STRING\",\n",
    "                \"street\": \"STRING\",\n",
    "                \"city\": \"STRING\",\n",
    "                \"state\": \"STRING\",\n",
    "                \"zip\": \"BIGINT\",\n",
    "                \"lat\": \"DOUBLE\",\n",
    "                \"long\": \"DOUBLE\",\n",
    "                \"city_pop\": \"BIGINT\",\n",
    "                \"job\": \"STRING\",\n",
    "                \"dob\": \"DATE\",\n",
    "            }\n",
    "        ),\n",
    "    )\n",
    "\n",
    "    # Feature query written in the Feldera SQL dialect.\n",
    "    query = \"\"\"\n",
    "        SELECT\n",
    "           t.cc_num,\n",
    "           -- Demographic attributes\n",
    "           zip,\n",
    "           city_pop,\n",
    "           -- Day-of-week\n",
    "           dayofweek(trans_date_trans_time) as d,\n",
    "           -- is_weekend flag\n",
    "           CASE\n",
    "             WHEN dayofweek(trans_date_trans_time) IN(6, 7) THEN true\n",
    "             ELSE false\n",
    "           END AS is_weekend,\n",
    "           -- hour of day\n",
    "           hour(trans_date_trans_time) as hour_of_day,\n",
    "           -- is_night flag\n",
    "           CASE\n",
    "             WHEN hour(trans_date_trans_time) <= 6 THEN true\n",
    "             ELSE false\n",
    "           END AS is_night,\n",
    "           -- Average spending per day, per week, and per month.\n",
    "           AVG(amt) OVER window_1_day AS avg_spend_pd,\n",
    "           AVG(amt) OVER window_7_day AS avg_spend_pw,\n",
    "           AVG(amt) OVER window_30_day AS avg_spend_pm,\n",
    "           -- Average spending over the last three months for the same day of the week.\n",
    "           COALESCE(\n",
    "            AVG(amt) OVER (\n",
    "              PARTITION BY t.cc_num, EXTRACT(DAY FROM trans_date_trans_time)\n",
    "              ORDER BY unix_time\n",
    "              RANGE BETWEEN 7776000 PRECEDING and CURRENT ROW\n",
    "            ), 0) AS avg_spend_p3m_over_d,\n",
    "           -- Number of transactions in the last 24 hours.\n",
    "           COUNT(*) OVER window_1_day AS trans_freq_24,\n",
    "           -- Transaction amount\n",
    "           amt,\n",
    "           -- Transaction time\n",
    "           unix_time,\n",
    "           -- Ground truth label\n",
    "           is_fraud\n",
    "        -- Enrich transaction data with demographic data\n",
    "        FROM transaction as t\n",
    "        JOIN demographics as d\n",
    "        ON t.cc_num = d.cc_num\n",
    "        WINDOW\n",
    "          window_1_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 86400 PRECEDING AND CURRENT ROW),\n",
    "          window_7_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 604800 PRECEDING AND CURRENT ROW),\n",
    "          window_30_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 2592000 PRECEDING AND CURRENT ROW);\n",
    "      \"\"\"\n",
    "\n",
    "    sql.register_view(\"FEATURE\", query)\n",
    "    return sql\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5b995f6",
   "metadata": {},
   "source": [
    "## Train & test the model\n",
    "\n",
    "Overview of the following code:\n",
    "- Connect to a Feldera service (we use [try.feldera.com](https://try.feldera.com))\n",
    "- Create a pipeline to evaluate feature queries over transaction and demographics tables in S3\n",
    "- Run the pipeline to process all input data **to completion**.\n",
    "- Read the computed features into a Pandas dataframe\n",
    "- Split the dataframe into train and test sets\n",
    "- Train an XGBoost model and measure its accuracy."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "16e0d435",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Helper functions for model training & testing\n",
    "\n",
    "from xgboost import XGBClassifier\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import confusion_matrix\n",
    "\n",
    "# Split input dataframe into train and test sets\n",
    "def get_train_test_data(dataframe, feature_cols, target_col, train_test_split_ratio, random_seed):\n",
    "    X = dataframe[feature_cols]\n",
    "    y = dataframe[target_col]\n",
    "    X_train, X_test, y_train, y_test = train_test_split(X, y, train_size = train_test_split_ratio, random_state = random_seed)\n",
    "\n",
    "    return X_train, X_test, y_train, y_test\n",
    "\n",
    "# Train a decision tree classifier using xgboost.\n",
    "# Other ML frameworks and types of ML models can be readily used with Feldera.\n",
    "def train_model(dataframe, config):\n",
    "    max_depth = 12\n",
    "    n_estimators = 100\n",
    "\n",
    "    X_train, X_test, y_train, y_test = get_train_test_data(\n",
    "        dataframe,\n",
    "        config['feature_cols'],\n",
    "        config['target_col'],\n",
    "        config['train_test_split_ratio'],\n",
    "        config['random_seed'])\n",
    "\n",
    "    model = XGBClassifier(\n",
    "        max_depth=max_depth,\n",
    "        n_estimators=n_estimators,\n",
    "        objective=\"binary:logistic\")\n",
    "\n",
    "    model.fit(X_train, y_train.values.ravel())\n",
    "    return model, X_test, y_test\n",
    "\n",
    "# Evaluate prediction accuracy against ground truth.\n",
    "def eval_metrics(y, predictions):\n",
    "    cm = confusion_matrix(y, predictions)\n",
    "    print(\"Confusion matrix:\")\n",
    "    print(cm)\n",
    "\n",
    "    if len(cm) < 2 or cm[1][1] == 0:  # checking if there are no true positives\n",
    "        print('No fraudulent transaction to evaluate')\n",
    "        return\n",
    "    else:\n",
    "        precision = cm[1][1] / (cm[1][1] + cm[0][1])\n",
    "        recall = cm[1][1] / (cm[1][1] + cm[1][0])\n",
    "        f1 = (2 * (precision * recall) / (precision + recall))\n",
    "\n",
    "    print(f\"Precision: {precision * 100:.2f}%\")\n",
    "    print(f\"Recall: {recall * 100:.2f}%\")\n",
    "    print(f\"F1 Score: {f1 * 100:.2f}%\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "17619d5d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "DATA_URI = \"s3://feldera-fraud-detection-data\"\n",
    "#DATA_URI = \"/home/leonid/projects/feldera/demo/project_demo10-FraudDetectionDeltaLake/data\"\n",
    "\n",
    "# Connect to the Feldera sandbox.\n",
    "# Use the 'Settings' menu at try.feldera.com to generate an API key\n",
    "client = FelderaClient(\"http://localhost:8080\")\n",
    "\n",
    "sql = build_program(client, \"fraud_detection_training\")\n",
    "\n",
    "# Load DEMOGRAPHICS data from a Delta table stored in a public S3 bucket.\n",
    "sql.connect_source_delta_table(\n",
    "    \"DEMOGRAPHICS\",\n",
    "    \"demographics_train\",\n",
    "    {\n",
    "        \"uri\": f\"{DATA_URI}/demographics_train/\",\n",
    "        \"mode\": \"snapshot\",\n",
    "        \"aws_skip_signature\": \"true\"\n",
    "    }\n",
    ")\n",
    "\n",
    "# Load credit card TRANSACTION data.\n",
    "sql.connect_source_delta_table(\n",
    "    \"TRANSACTION\",\n",
    "    \"transaction_train\",\n",
    "    {\n",
    "        \"uri\": f\"{DATA_URI}/transaction_train/\",\n",
    "        \"mode\": \"snapshot\",\n",
    "        \"aws_skip_signature\": \"true\",\n",
    "        \"timestamp_column\": \"unix_time\"\n",
    "    }\n",
    ")\n",
    "\n",
    "# sql.connect_sink_delta_table(\n",
    "#     \"FEATURE\",\n",
    "#     \"feature_train\",\n",
    "#     {\n",
    "#         \"uri\": \"s3://feldera-fraud-detection-demo/feature_train\",\n",
    "#         \"mode\": \"truncate\",\n",
    "#         \"aws_access_key_id\": dbutils.secrets.get(\"feature-engineering-demo\", \"AWS_ACCESS_KEY_ID\"),\n",
    "#         \"aws_secret_access_key\": dbutils.secrets.get(\"feature-engineering-demo\", \"AWS_SECRET_ACCESS_KEY\"),\n",
    "#         \"aws_region\": \"us-east-1\"\n",
    "\n",
    "#     }\n",
    "# )\n",
    "\n",
    "hfeature = sql.listen(\"feature\")\n",
    "\n",
    "# Process full snapshot of the input tables and compute a dataset with feature vectors.\n",
    "sql.start()\n",
    "sql.wait_for_completion(shutdown=True)\n",
    "\n",
    "# Read computed feature vectors into a Pandas dataframe.\n",
    "features_pd = hfeature.to_pandas()\n",
    "print(f\"Computed {len(features_pd)} feature vectors\")\n",
    "\n",
    "print(\"Training the model\")\n",
    "\n",
    "feature_cols = list(features_pd.columns.drop('is_fraud'))\n",
    "\n",
    "config={\n",
    "        'feature_cols' : feature_cols,\n",
    "        'target_col' : ['is_fraud'],\n",
    "        'random_seed' : 45,\n",
    "        'train_test_split_ratio' : 0.8\n",
    "        }\n",
    "\n",
    "trained_model, X_test, y_test = train_model(features_pd, config)\n",
    "\n",
    "print(\"Testing the trained model\")\n",
    "\n",
    "y_pred = trained_model.predict(X_test)\n",
    "eval_metrics(y_test, y_pred)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "86693081",
   "metadata": {},
   "source": [
    "## REAL-TIME INFERENCE\n",
    "\n",
    "During real-time feature computation, raw data arrives from a streaming source like Kafka. Feldera can ingest data directly from such sources, but in this case we will assume that Kafka is connected to a Delta table, and configure Feldera to ingest the data by following the transaction log of the table.\n",
    "\n",
    "<img src=\"inference_animation.gif\" alt=\"inference\" width=\"800\"/>\n",
    "\n",
    "Below, we create another Feldera pipeline to evaluate the feature query over streaming data:\n",
    "- Build a pipeline identical to the training pipeline above, but using the inference dataset as input\n",
    "- Configure the input connector for the `TRANSACTION` table to ingest transaction data in the `snapshot_and_follow` mode. In this mode, the connector reads the initial snapshot of the table before following the stream of changes in its transaction log. This **backfill** pattern is necessary to correctly evaluate features that depend on historical data such as rolling sums and averages.\n",
    "- Run the pipeline for 30 seconds. For each batch of new feature vectors computed by the pipeline:\n",
    "  - Read the data into a Pandas dataframe\n",
    "  - Feed the dataframe to the trained ML model for inference\n",
    "  - Measure model accuracy by comparing model prediction with the ground truth"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f77596ce",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Helper function: feed a Pandas dataframe to the trained model for inference.\n",
    "def inference(trained_model, df):\n",
    "    #print(f\"\\nReceived {len(df)} feature vectors.\")\n",
    "    if len(df) == 0:\n",
    "        return\n",
    "\n",
    "    feature_cols_inf = list(df.columns.drop('is_fraud'))\n",
    "    X_inf = df[feature_cols_inf].values  # convert to numpy array\n",
    "    y_inf = df[\"is_fraud\"].values\n",
    "    predictions_inf = trained_model.predict(X_inf)\n",
    "\n",
    "    # eval_metrics(y_inf, predictions_inf)\n",
    "    fraud = [index for index, value in enumerate(predictions_inf) if value != 0]\n",
    "\n",
    "    GREEN = \"\\033[92m\"\n",
    "    RED = \"\\033[91m\"\n",
    "    RESET = \"\\033[0m\"\n",
    "\n",
    "    if len(fraud) > 0:\n",
    "        print(f\"{RED}Detected {len(fraud)} fraudulent transactions (out of {len(df)}){RESET}\")\n",
    "        #print(f\"{RED}Example:\\n{df.iloc[[fraud[0]]]}{RESET}\")\n",
    "    else:\n",
    "        print(f\"{GREEN}No fraudulent transaction detected{RESET}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7618c8c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "# How long to run the inference pipeline for.\n",
    "INFERENCE_TIME_SECONDS = 30\n",
    "\n",
    "print(f\"Running the inference pipeline\")\n",
    "\n",
    "sql = build_program(client, \"fraud_detection_inference\")\n",
    "\n",
    "# Load DEMOGRAPHICS data from a Delta table.\n",
    "sql.connect_source_delta_table(\n",
    "    \"DEMOGRAPHICS\",\n",
    "    \"demographics_infer\",\n",
    "    {\n",
    "        \"uri\": f\"{DATA_URI}/demographics_infer\",\n",
    "        \"mode\": \"snapshot\",\n",
    "        \"aws_skip_signature\": \"true\"\n",
    "    }\n",
    ")\n",
    "\n",
    "# Read TRANSACTION data from a Delta table.\n",
    "# Configure the Delta Lake connector to read the initial snapshot of\n",
    "# the table before following the stream of changes in its transaction log.\n",
    "sql.connect_source_delta_table(\n",
    "    \"TRANSACTION\",\n",
    "    \"transaction_infer\",\n",
    "    {\n",
    "        \"uri\": f\"{DATA_URI}/transaction_infer\",\n",
    "        \"mode\": \"snapshot_and_follow\",\n",
    "        \"version\": 10,\n",
    "        \"timestamp_column\": \"unix_time\",\n",
    "        \"aws_skip_signature\": \"true\"\n",
    "    }\n",
    ")\n",
    "\n",
    "# sql.connect_sink_delta_table(\n",
    "#     \"FEATURE\",\n",
    "#     \"feature_infer\",\n",
    "#     {\n",
    "#         \"uri\": \"s3://feldera-fraud-detection-demo/feature_infer\",\n",
    "#         \"mode\": \"truncate\",\n",
    "#         \"aws_access_key_id\": dbutils.secrets.get(\"feature-engineering-demo\", \"AWS_ACCESS_KEY_ID\"),\n",
    "#         \"aws_secret_access_key\": dbutils.secrets.get(\"feature-engineering-demo\", \"AWS_SECRET_ACCESS_KEY\"),\n",
    "#         \"aws_region\": \"us-east-1\"\n",
    "\n",
    "#     }\n",
    "# )\n",
    "\n",
    "sql.foreach_chunk(\"feature\", lambda df, chunk : inference(trained_model, df))\n",
    "\n",
    "# Start the pipeline to continuously process the input stream of credit card\n",
    "# transactions and output newly computed feature vectors to a Delta table.\n",
    "sql.start()\n",
    "\n",
    "time.sleep(INFERENCE_TIME_SECONDS)\n",
    "\n",
    "print(f\"Shutting down the inference pipeline after {INFERENCE_TIME_SECONDS} seconds\")\n",
    "sql.shutdown()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bc778df9",
   "metadata": {},
   "source": [
    "### Monitoring the inference pipeline in the Web Console\n",
    "\n",
    "While the above code is running, you can monitor the pipeline in the Feldera Web Console:\n",
    "\n",
    "<img src=\"web_console.gif\" alt = \"Web Console\" width=\"1000\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3c10a16b",
   "metadata": {},
   "source": [
    "## TAKEAWAYS\n",
    "\n",
    "In this example we used Feldera to evaluate *the same feature queries* first over historical (batch) data and then over a combination of historical and streaming inputs.  Feldera's ability to operate on any combination of batch and streaming sources eliminates the need to develop multiple implementations of the same queries for development and production environments.  **In fact, Feldera does not distinguish between the two**.  It processes inputs in the same way and produces the same outputs, whether they arrive frequently in small groups (aka streaming) or occasionally in bigger groups (aka batch).\n",
    "\n",
    "Upon receiving new inputs, Feldera updates its output views without full re-computation, by doing work proportional to the size of the new data rather than the size of the entire database.  This **incremental evaluation** makes Feldera efficient for both streaming and batch inputs.\n",
    "\n",
    "Finally, we would like to emphaize that **Feldera is strongly consistent**. If\n",
    "we pause our inference pipeline and inspect the contents of the output view\n",
    "produced by Feldera so far, it will be **precisely the same as if we ran the\n",
    "query on all the inputs received so far as one large batch**.  Unpause the\n",
    "pipeline and run it a little longer.  The pipeline will receive some additional\n",
    "inputs and produce additional outputs, but it still preserves the same\n",
    "input/output guarantee.  This property, known as **strong consistency**, ensures\n",
    "that the prediction accuracy of your ML model will not be affected by incorrect input.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "76c32680",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
